dbtでSlowly Changing Dimensions Type2を用いたテーブルの変更履歴の保持が簡単に実装できるsnapshotを試してみた #dbt

モダンデータスタック(MDS)

#dbt

さがら

2022.06.23

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

さがらです。

dbtでSlowly Changing Dimensions Type2を用いたテーブルの変更履歴の保持が簡単に実装できるsnapshotという機能を試してみたので、本記事でまとめてみます。

Slowly Changing Dimensions Type2とは

まず、タイトルにもあるSlowly Changing Dimensions Type2とは何か、説明していきます。

Slowly changing dimensionとは

一言でいうと、「Slowly Changing Dimensionsは、ディメンションテーブルの変更履歴を管理する手法」です。

まず大前提としてデータ分析をする上で、ディメンションという概念があり、データを分析する際に切り口となるデータのことを示します。ディメンションテーブルなどと呼ばれますね。

このディメンションテーブルの例としては

取り扱う商品の詳細情報（品名、価格、など）を集めたテーブル
あるECサイトの会員情報（氏名、住所、など）を集めたテーブル

といったデータが該当します。

その上で、こういったテーブルは不定期にデータの内容がアップデートされることが多いです。具体的には、下記のような変更が行われます。

取り扱う商品の詳細情報（品名、価格、など）を集めたテーブル
- 原材料の単価上昇による価格変更
- 商流変更による入荷元の変更
あるECサイトの会員情報（氏名、住所、など）を集めたテーブル
- 会員の引っ越しによる住所変更
- 購入状況に合わせた会員ランクの変更

こういった変更は毎日頻繁に行われるものではなく、不定期に別の値に切り替わります。

そしてこの上で、「ある年月日時点の会員情報を知りたい」という場面は実務上多く発生します。そのため、対象のディメンションテーブルの変更履歴を管理する必要があり、その変更履歴の管理手法がSlowly Changing Dimensionsとして確立されている、ということに繋がってきます。

Type2とは

続いて末尾のType2なのですが、Slowly Changing DimensionsについてはType0～Type7まであり、そのうちの1つの手法を意味しています。（他の手法については、Wikipediaを参照してみてください。）

Type2は簡単にいうと、対象のディメンションテーブル上に新しい行を追加して変更履歴を管理する手法です。

以下、dbtのsnapshotのドキュメントの画像を引用して、どのような挙動となるか確認していきます。

まず、管理する商品の出荷状態を管理するordersというテーブルがあると仮定します。このテーブルの中身が、下図のようにあるidの商品がまだ配送前の状態でstatusカラムの値がpendingだったとします。

このとき、このidの商品が出荷され、statusカラムの値がshippedに変更されたとします。

この変更が行われた時にSlowly Changing Dimensions Type2のロジックを実装していると、下図のように、statusがpendingとshippedの時で行が分かれ、いつからいつまでがこの値であったかを示すことが出来るようになります。

このように、変更履歴を新しい行の追加により管理していくロジックを実装するのが、Slowly Changing Dimensions Type2となります。

dbtのsnapshotとは

前置きが少し長くなりましたが、dbtのsnapshotは前述のSlowly Changing Dimensions Type2を実装できる機能です。

公式Docからの引用ですが、設定の際は下記のように記述します。

{% snapshot orders_snapshot %}

{{
    config(
      target_database='analytics',
      target_schema='snapshots',
      unique_key='id',

      strategy='timestamp',
      updated_at='updated_at',
    )
}}

select * from {{ source('jaffle_shop', 'orders') }}

{% endsnapshot %}

以下、snapshotを定義する際の基本的なパラメータについて説明します。

target_database
- オプションの設定です。snapshotの生成先となるデータベース名を指定します。
target_schema
- 必須の設定です。snapshotの生成先となるスキーマ名を指定します。
unique_key
- 必須の設定です。変更履歴を追う際の基準となる主キーを指定します。
strategy
- 「timestamp」と「check」の2種類から選択します。選択するstrategyによって、後述のパラメータで必須の物が変わってきます。
- dbtとしては、「timestamp」を使用することを推奨しています。
updated_at
- 上述のstrategyが「timestamp」の場合に必須となる設定です。
- 対象のテーブルで、レコードが更新されたことを確認できるカラム名を入力します。
check_cols
- 上述のstrategyが「check」の場合に必須となる設定です。
- このパラメータで指定したカラムの内容が、対象のテーブルにおいて更新されたらそのレコードの変更履歴を追うことができます。
- 入力例：["name", "email"]

snapshotに関するベストプラクティス

snapshotはとても便利な機能なのですが、使い方を誤ると、全く活用されない大量のデータを保持してしまうなど、リスクに繋がる可能性があります。

以下は公式Docからの引用ですが、snapshotに関するベストプラクティスを、私なりに解釈してまとめておきます。

出来る限り、strategyは「timestamp」を採用する
設定するunique_keyが本当にユニークであることを確認する
- uniqueであるかを確認するschema testを設定するのもオススメ
snapshot用に別のスキーマを定義する
- 分析用のデータを保持するスキーマと同じ階層にsnapshotを生成すると、意図せぬ形でsnapshotが削除されたときに再構築できません
- 例えばSnowflakeの場合、snapshot生成のときだけ別のロールを設定して、意図せぬ削除を防ぐなどの運用もオススメ
sourceにあたるデータに対してsnapshotを行う
- クリーニング処理を行う前の生データに対してsnapshotを取るのが良い
sourceに当たるデータを参照するときはsource関数を使う
- これにより、sourceからのデータリネージを確認できる
snapshotには、対象のテーブルからなるべく多くのカラムを選択するようにしましょう
- 理想はselect *で全てのカラムを選択する。今は使用していないカラムでも、将来的に役立つ可能性がある
snapshotに設定するSELCT文では、JOINを避ける
- snapshotの更新に用いるtimestampのカラムの信用性が取れなくなる可能性が高いため
- もしJOINが必要な場合は、元のテーブルそれぞれでsnapshotを取ること
snapshotに設定するSELECT文では、データのクリーニング処理やロジックの実装など、変換処理を避ける
- そのクリーニング処理や実装したロジックが、いつ変更が必要となるかわからない
- 基本的にはシンプルなクエリを用いてsnapshotを取ること。例外的なパターンとしては、JSONで保持されているカラムを展開してsnapshotを取っておきたい場合などがある

snapshot試してみた

実際に試していきたいと思います。dbtとしても推奨事項である「Timestamp strategy」を用いてみます。

対象のデータ

下図のようなデータを使います。LAST_UPDATEDカラムを用いてレコードの変化を検知していきます。

また、dbtのベストプラクティスに沿って、事前にdbt上でこのデータに対するsourceの設定もしておきます。

snapshotの実装

続いて、dbt上でsnapshotを実装していきます。

デフォルトのプロジェクト構成で、snapshotsというフォルダがあるため、その中にsqlファイルを新しく作成します。

作成したsqlファイルの中に、以下のコードを記述し、保存します。

{% snapshot snapshot_test_table %}

    {{
        config(
          target_schema='snapshots',
          strategy='timestamp',
          unique_key='uuid',
          updated_at='last_updated',
        )
    }}

    select * from {{ source('snapshot_test','FOR_DBT_SNAPSHOT_TEST') }}

{% endsnapshot %}

snapshot用のコードの記述が完了したら、dbt snapshotコマンドを実行します。

このコマンドが成功すると、dbt用のデータベース内のtarget_schemaで指定したスキーマ上に、{% snapshot snapshot_test_table %}と指定したテーブル名で、snapshotが作られました。

実際に中身を見ると、このようなテーブルとなっています。元のテーブルでは3列だけでしたが、snapshotを取ったことにより4列追加されていますね。

追加されたカラムの内容について、下記にまとめます。

DBT_SCD_ID
- snapshotされた各レコードに対して生成される一意のキー。dbtがsnapshotの内部処理で使用するキーです
DBT_UPDATED_AT
- snapshot元のテーブルにおける、updated_atに設定されたカラムの値(タイムスタンプ)。dbtがsnapshotの内部処理で使用します
DBT_VALID_FROM
- 対象のレコードが有効になった時のタイムスタンプ
DBT_VALID_TO
- 対象のレコードが無効になった時のタイムスタンプ
- 対象のレコードが、「最新のレコード」あるいは「初めて登録されたレコード」の場合は何も入力されません

snapshotの更新

続いて、一度データの内容をUPDATEした上で、どのようにsnapshotの内容が変わるのか見ていきます。

下記のSQLを実行してみます。

update SAGARA_RAWDATA_DB.PUBLIC.FOR_DBT_SNAPSHOT_TEST
    set transportation_type = 'air',
        last_updated = current_timestamp
    where uuid = '5cc0bf01-a22c-4223-94e3-1ba33ac4e109';